#include "znx_worker_proc.h"
#include "znx_event_loop.h"

char *znx_worker_thread_tag = "worker_thread_tag";

znx_worker_processor_t          *g_worker_processor;
static int64_t                  worker_thread_idx = 0;


void
znx_worker_processor_init()
{
    g_worker_processor = znx_thread_calloc(sizeof(znx_worker_processor_t));
    g_worker_processor->init_hook_chain = znx_hooks_create();
    g_worker_processor->proc_exit_chain = znx_hooks_create();
    g_worker_processor->quit_hook_chain = znx_hooks_create();
}


static inline void
znx_worker_event_loop_init(znx_event_loop_data_t *loop_data)
{
    if (!znx_hooks_exec(g_worker_processor->init_hook_chain, loop_data)) {
        ZNX_DEF_LOG_FATAL("worker event loop init failed");
    }
}


static inline void
znx_worker_event_loop_proc_exit(znx_event_loop_data_t *loop_data)
{
    if (!znx_hooks_exec(g_worker_processor->proc_exit_chain, loop_data)) {
        ZNX_DEF_LOG_ERROR("worker event proc exit failed");
    }
}


static inline void
znx_worker_event_loop_quit(znx_event_loop_data_t *loop_data)
{
    if (!znx_hooks_exec(g_worker_processor->quit_hook_chain, loop_data)) {
        ZNX_DEF_LOG_ERROR("worker event quit failed");
    }
}


// hook_add must be called before znx_worker_processor_start.
void
znx_worker_processor_add_init_hook(znx_hook_handler_pt hook_handler)
{
    znx_hooks_add(g_worker_processor->init_hook_chain, hook_handler);
}


// hook_add must be called before znx_worker_processor_start.
void
znx_worker_processor_add_proc_exit_hook(znx_hook_handler_pt hook_handler)
{
    znx_hooks_add(g_worker_processor->proc_exit_chain, hook_handler);
}


// hook_add must be called before znx_worker_processor_start.
void
znx_worker_processor_add_quit_hook(znx_hook_handler_pt hook_handler)
{
    znx_hooks_add(g_worker_processor->quit_hook_chain, hook_handler);
}


// run in main thread.
void
znx_worker_processor_start(
    int nthreads,
    int64_t mem_freelist_level,
    int64_t aligned_mem_freelists_level)
{
    if (nthreads <= 0) {
        ZNX_DEF_LOG_FATAL("worker threads must be greate than 0");
    }

    znx_event_loop_data_t   *datas;
    znx_thread_base_t       *threads;

    datas = znx_thread_calloc(
        sizeof(znx_event_loop_data_t) * (size_t)nthreads);
    threads = znx_thread_calloc(
        sizeof(znx_thread_base_t) * (size_t)nthreads);

    char thread_name[64] = {0};

    for (int i = 0; i < nthreads; i++) {
        snprintf(thread_name, 64, "worker-%d", i);

        znx_event_loop_data_init(
            datas+i,
            znx_worker_event_loop_init,
            znx_worker_event_loop_proc_exit,
            znx_worker_event_loop_quit);

        znx_thread_base_start(
            znx_worker_thread_tag,
            threads+i,
            znx_event_loop_runner,
            datas+i,
            mem_freelist_level,
            aligned_mem_freelists_level
        );

        znx_thread_set_name((threads+i)->tid, thread_name);
    }

    g_worker_processor->nthreads = nthreads;
    g_worker_processor->threads = threads;
    g_worker_processor->datas = datas;
}


void
znx_worker_processor_stop()
{
    for (int i = 0; i < g_worker_processor->nthreads; i++) {
        znx_event_loop_quit(g_worker_processor->threads+i);
    }

    for (int i = 0; i < g_worker_processor->nthreads; i++) {
        znx_thread_base_join(g_worker_processor->threads+i, NULL);
    }

    znx_thread_free(g_worker_processor->datas);
    znx_thread_free(g_worker_processor->threads);
}


void
znx_worker_processor_post_event(znx_dispatch_event_t *dispatch_event)
{
    if (g_worker_processor->nthreads == 0) {
        ZNX_DEF_LOG_FATAL("worker threads must be greate than 0");
    }

    worker_thread_idx = (worker_thread_idx + 1) % g_worker_processor->nthreads;
    znx_event_loop_post_event(
        g_worker_processor->threads + worker_thread_idx,
        dispatch_event);
}


znx_bool_t
znx_worker_processor_all_task_done()
{
    znx_event_loop_data_t   *data;
    for (int i = 0; i < g_worker_processor->nthreads; i++) {
        data = g_worker_processor->datas + i;
        if (!znx_atomic_bool_is_true(&data->can_stop)) {
            return ZNX_FALSE;
        }
    }

    return ZNX_TRUE;
}
